package com.cskaoyan.user.mq;

import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.mail.SimpleMailMessage;
import org.springframework.mail.javamail.JavaMailSender;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.nio.charset.StandardCharsets;
import java.util.List;

/**
 * @program: csmall-practice
 * @description: 发送邮件的消费者
 * @author: Lexis
 * @create: 2022-05-23 22:26
 **/

@Slf4j
@Component
public class SendEmailConsumer {

    @Autowired
    JavaMailSender mailSender;

    private DefaultMQPushConsumer consumer;


    @PostConstruct
    public void init() {
        consumer = new DefaultMQPushConsumer("send_email_consumer");

        consumer.setNamesrvAddr("127.0.0.1:9876");

        consumer.setMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                try {
                    MessageExt message = list.get(0);

                    String from = message.getUserProperty("from");
                    String to = message.getUserProperty("to");
                    String subject = message.getUserProperty("subject");
                    byte[] body = message.getBody();
                    // 解析并消费
                    String text = new String(body, 0, body.length, StandardCharsets.UTF_8);


                    SimpleMailMessage sendEmail = new SimpleMailMessage();
                    sendEmail.setSubject(subject);
                    sendEmail.setFrom(from);
                    sendEmail.setTo(to);
                    sendEmail.setText(text);
                    mailSender.send(sendEmail);
                    log.info("邮件发送成功");
                } catch (Exception e) {
                    e.printStackTrace();
                    return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;

            }
        });

        try {
            consumer.subscribe("send_email", "*");
            consumer.start();
        } catch (MQClientException e) {
            e.printStackTrace();
        }
    }
}

